package com.example.flow;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

public class FlowDemo {

    public static void main(String[] args) {

        /**
         * 发布者：上游
         */
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        /**
         * 订阅者：下游
         */
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {

            /**
             * 订阅关系
             */
            private Flow.Subscription subscription;

            /**
             * 订阅数量
             */
            private int count = 0;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                this.subscription.request(5);
            }

            /**
             * 在 Java Flow API 里，即使你调用了：
             * subscription.cancel();
             * 发布者端那边可能还有已经排队但尚未投递到 onNext 的数据项。
             * 因为 SubmissionPublisher 的工作方式是：
             * 提交数据（submit）后，数据会被放进一个内部的无界/有界队列；
             * 一个异步任务从队列中取出数据，然后回调订阅者的 onNext。
             * 这就有一个竞态条件（race condition）：
             * 你在 onNext 中调用 cancel()；
             * 发布者后台线程可能已经把下一个数据从队列里取出来，准备调用 onNext；
             * 虽然订阅关系已经标记为取消，但这次回调依然会被调用（具体取决于实现，SubmissionPublisher 就是可能的）。
             * 没有标志位的话，这些多余的回调就会继续执行你的处理逻辑（打印、计数、甚至 request），和你的“取消”语义冲突。
             */
            @Override
            public void onNext(String item) {
                if (count == 38) {
                    /**
                     * 取消订阅
                     */
                    this.subscription.cancel();
                }
                System.out.println("订阅数据：" + item);
                if (++count % 5 == 0) {
                    this.subscription.request(5);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("订阅数据异常：" + throwable);
            }

            @Override
            public void onComplete() {
                System.out.println("订阅数据完成");
            }
        };

        /**
         * 绑定订阅者与发布者
         */
        publisher.subscribe(subscriber);

        /**
         * 发布数据
         */
        for (int i = 0; i < 50; i++) {
            publisher.submit("发布数据：" + i);
        }

        /**
         * 关闭发布者
         */
        publisher.close();

        /**
         * 休眠1秒，等待数据处理完成
         */
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        }

    }

}
